 1.HDFS
   
   该引擎提供了集成 Apache Hadoop 生态系统通过允许管理数据 HDFS通过ClickHouse. 这个引擎是相似的 到 文件
和 URL 引擎，但提供Hadoop特定的功能。
   用途
   ENGINE = HDFS(URI, format)
   该 URI 参数是HDFS中的整个文件URI。 该 format 参数指定一种可用的文件格式。 执行 SELECT 查询时，格式必须
支持输入，并执行 INSERT queries – for output. The available formats are listed in the 格式 科。 路径部分 URI 可
能包含水珠。 在这种情况下，表将是只读的。
   示例:
   1. 设置 hdfs_engine_table 表:
CREATE TABLE hdfs_engine_table (name String, value UInt32)
ENGINE=HDFS('hdfs://hdfs1:9000/other_storage', 'TSV')
   2. 填充文件:
INSERT INTO hdfs_engine_table VALUES ('one', 1), ('two', 2), ('three', 3);
   3. 查询数据:
SELECT * FROM hdfs_engine_table LIMIT 2;
┌─name─┬─value─┐
│ one │ 1 │
│ two │ 2 │
└──────┴───────┘   
   
   实施细节
       读取和写入可以并行
       不支持:
           ALTER 和 SELECT...SAMPLE 操作。
           索引。
           复制。
   路径中的水珠
   多个路径组件可以具有globs。 对于正在处理的文件应该存在并匹配到整个路径模式。 文件列表确定在 SELECT （不
在 CREATE 时刻）。
       — Substitutes any number of any characters except / 包括空字符串。
       ? — Substitutes any single character.
       {some_string,another_string,yet_another_one} — Substitutes any of strings 'some_string',
'another_string', 'yet_another_one'.
       {N..M} — Substitutes any number in range from N to M including both borders.
   建筑与 {} 类似于 远程 表功能。
   示例
       1. 假设我们在HDFS上有几个TSV格式的文件，其中包含以下Uri:
		  'hdfs://hdfs1:9000/some_dir/some_file_1'
		  'hdfs://hdfs1:9000/some_dir/some_file_2'
		  'hdfs://hdfs1:9000/some_dir/some_file_3'
		  'hdfs://hdfs1:9000/another_dir/some_file_1'
		  'hdfs://hdfs1:9000/another_dir/some_file_2'
		  'hdfs://hdfs1:9000/another_dir/some_file_3'
       
	   1. 有几种方法可以创建由所有六个文件组成的表:
CREATE TABLE table_with_range (name String, value UInt32) ENGINE =
HDFS('hdfs://hdfs1:9000/{some,another}_dir/some_file_{1..3}', 'TSV')

   另一种方式:
CREATE TABLE table_with_question_mark (name String, value UInt32) ENGINE =
HDFS('hdfs://hdfs1:9000/{some,another}_dir/some_file_?', 'TSV')
   表由两个目录中的所有文件组成（所有文件都应满足query中描述的格式和模式):
CREATE TABLE table_with_asterisk (name String, value UInt32) ENGINE =
HDFS('hdfs://hdfs1:9000/{some,another}_dir/*', 'TSV')
   警告
   如果文件列表包含带有前导零的数字范围，请单独使用带有大括号的构造或使用 ?.
   示例
   创建具有名为文件的表 file000, file001, … , file999: 
CREARE TABLE big_table (name String, value UInt32) ENGINE =
HDFS('hdfs://hdfs1:9000/big_dir/file{0..9}{0..9}{0..9}', 'CSV')
   虚拟列
       _path — Path to the file.
       _file — Name of the file.
   另请参阅
       虚拟列

 2.MySQL
   
   MySQL 引擎可以对存储在远程 MySQL 服务器上的数据执行 SELECT 查询。
   调用格式：
MySQL('host:port', 'database', 'table', 'user', 'password'[, replace_query,
'on_duplicate_clause']);
   调用参数
       host:port — MySQL 服务器地址。
	   database — 数据库的名称。
	   table — 表名称。
	   user — 数据库用户。
	   password — 用户密码。   
	   replace_query — 将 INSERT INTO 查询是否替换为 REPLACE INTO 的标志。如果replace_query=1,则替换查询
       'on_duplicate_clause' — 将 ON DUPLICATE KEY UPDATE 'on_duplicate_clause' 表达式添加到INSERT查询语
句中。例如：impression = VALUES(impression) + impression。如果需要指定 'on_duplicate_clause',则需要设置
replace_query=0. 如果同时设置 replace_query = 1 和 'on_duplicate_clause'，则会抛出异常
   
   此时，简单的 WHERE 子句（例如 =, !=, >, >=, <, <=）是在 MySQL 服务器上执行。
   其余条件以及 LIMIT 采样约束语句仅在对MySQL的查询完成后才在ClickHouse中执行。
   MySQL 引擎不支持 可为空 数据类型，因此，当从MySQL表中读取数据时，NULL 将转换为指定列类型的默认值（通
常为0或空字符串）。
   案例：
CREATE TABLE mysql_table
(
`id` UInt32,
`name` String,
`age` UInt32
)
ENGINE = MySQL('192.168.71.102:3306', 'bigdata', 'student', 'root', '12345678')

   SELECT *
FROM mysql_table2;

┌─id─┬─name──┬─age─┐
│ 3 │ jack │ 28 │
│ 4 │ lucas │ 18 │
└────┴───────┴─────┘ 

 3.Kafka
   
   此引擎与 Apache Kafka 结合使用。
   Kafka 特性：
       发布或者订阅数据流。
       容错存储机制。
       处理流数据。
   老版格式：
Kafka(kafka_broker_list, kafka_topic_list, kafka_group_name, kafka_format
[, kafka_row_delimiter, kafka_schema, kafka_num_consumers]);
   新版格式：
Kafka SETTINGS
  kafka_broker_list = 'localhost:9092',
  kafka_topic_list = 'topic1,topic2',
  kafka_group_name = 'group1',
  kafka_format = 'JSONEachRow',
  kafka_row_delimiter = '\n',
  kafka_schema = '',
  kafka_num_consumers = 2
    
   必要参数：
       kafka_broker_list – 以逗号分隔的 brokers 列表 (localhost:9092)。
       kafka_topic_list – topic 列表 (my_topic)。
	   kafka_group_name – Kafka 消费组名称 (group1)。如果不希望消息在集群中重复，请在每个分片中使用相同
的组名
       kafka_format – 消息体格式。使用与 SQL 部分的 FORMAT 函数相同表示方法，例如 JSONEachRow。了解详细
信息，请参考 Formats 部分。
   
   可选参数：
       kafka_row_delimiter - 每个消息体（记录）之间的分隔符。
	   kafka_schema – 如果解析格式需要一个 schema 时，此参数必填。例如，普罗托船长 需要 schema 文件路径
以及根对象 schema.capnp:Message 的名字。
       kafka_num_consumers – 单个表的消费者数量。默认值是：1，如果一个消费者的吞吐量不足，则指定更多的
消费者。消费者的总数不应该超过 topic 中分区的数量，因为每个分区只能分配一个消费者。

   示例：
CREATE TABLE queue (
timestamp UInt64,
level String,
message String
) ENGINE = Kafka('localhost:9092', 'topic', 'group1', 'JSONEachRow');

SELECT * FROM queue LIMIT 5;

CREATE TABLE queue2 (
timestamp UInt64,
level String,
message String
) ENGINE = Kafka SETTINGS kafka_broker_list = 'linux121:9092',
kafka_topic_list = 'topic',
kafka_group_name = 'group1',
kafka_format = 'JSONEachRow',
kafka_num_consumers = 4;

CREATE TABLE queue2 (
timestamp UInt64,
level String,
message String
) ENGINE = Kafka('localhost:9092', 'topic', 'group1')
SETTINGS kafka_format = 'JSONEachRow',
kafka_num_consumers = 4;
   
   消费的消息会被自动追踪，因此每个消息在不同的消费组里只会记录一次。如果希望获得两次数据，则使用另一个组
名创建副本。
   消费组可以灵活配置并且在集群之间同步。例如，如果群集中有10个主题和5个表副本，则每个副本将获得2个主
题。如果副本数量发生变化，主题将自动在副本中重新分配。了解更多信息请访问 http://kafka.apache.org/intro。
   SELECT 查询对于读取消息并不是很有用（调试除外），因为每条消息只能被读取一次。使用物化视图创建实时线程
更实用。您可以这样做：
        1. 使用引擎创建一个 Kafka 消费者并作为一条数据流。
        2. 创建一个结构表。
        3. 创建物化视图，改视图会在后台转换引擎中的数据并将其放入之前创建的表中。
   当 MATERIALIZED VIEW 添加至引擎，它将会在后台收集数据。可以持续不断地从 Kafka 收集数据并通过 SELECT 将
数据转换为所需要的格式。
   示例：
CREATE TABLE queue (
timestamp UInt64,
level String,
message String
 ) ENGINE = Kafka('localhost:9092', 'topic', 'group1', 'JSONEachRow');



CREATE MATERIALIZED VIEW consumer TO daily
AS SELECT toDate(toDateTime(timestamp)) AS day, level, count() as total
FROM queue GROUP BY day, level;

CREATE TABLE queue (
q_date String,
level String,
message String
) ENGINE = Kafka SETTINGS kafka_broker_list = 'linux121:9092',
kafka_topic_list = 'ch2',
kafka_group_name = 'group1',
kafka_format = 'CSV',
kafka_num_consumers = 1,
kafka_skip_broken_messages =10;

CREATE TABLE daily (
day Date,
level String,
message String
) ENGINE = MergeTree(day, (day, level), 8192);

CREATE MATERIALIZED VIEW consumer TO daily
as select q_date , level, message
from queue;


SELECT level, sum(total) FROM daily GROUP BY level;   

   为了提高性能，接受的消息被分组为 max_insert_block_size 大小的块。如果未在 stream_flush_interval_ms 毫秒内
形成块，则不关心块的完整性，都会将数据刷新到表中。
   停止接收主题数据或更改转换逻辑，请 detach 物化视图：
DETACH TABLE consumer;
ATTACH TABLE consumer;

   如果使用 ALTER 更改目标表，为了避免目标表与视图中的数据之间存在差异，推荐停止物化视图。
   配置
   与 GraphiteMergeTree 类似，Kafka 引擎支持使用ClickHouse配置文件进行扩展配置。可以使用两个配置键：全局
(kafka) 和 主题级别 (kafka_*)。首先应用全局配置，然后应用主题级配置（如果存在）。

<!-- Global configuration options for all tables of Kafka engine type -->
<kafka>
	<debug>cgrp</debug>
	<auto_offset_reset>smallest</auto_offset_reset>
</kafka>

<!-- Configuration specific for topic "logs" -->
<kafka_logs>
	<retry_backoff_ms>250</retry_backoff_ms>
	<fetch_min_bytes>100000</fetch_min_bytes>
</kafka_logs

   有关详细配置选项列表，请参阅 librdkafka配置参考。在 ClickHouse 配置中使用下划线 (_) ，并不是使用点 (.)。例
如，check.crcs=true 将是 true。

 4.JDBC
   
   允许CH通过 JDBC 连接到外部数据库。
   要实现JDBC连接，CH需要使用以后台进程运行的程序 clickhouse-jdbc-bridge。
   该引擎支持 Nullable 数据类型。
   建表
CREATE TABLE [IF NOT EXISTS] [db.]table_name
(
columns list...
)
ENGINE = JDBC(dbms_uri, external_database, external_table)

   引擎参数
       dbms_uri — 外部DBMS的uri.
       格式: jdbc:://:/?user=&password=. MySQL示例: jdbc:mysql://localhost:3306/?user=root&password=root.
       external_database — 外部DBMS的数据库名.
       external_table — external_database中的外部表名.  
   用法示例
   通过mysql控制台客户端来创建表
   Creating a table in MySQL server by connecting directly with it’s console client:
mysql> CREATE TABLE `test`.`test` (
-> `int_id` INT NOT NULL AUTO_INCREMENT,
-> `int_nullable` INT NULL DEFAULT NULL,
-> `float` FLOAT NOT NULL,
-> `float_nullable` FLOAT NULL DEFAULT NULL,
-> PRIMARY KEY (`int_id`));
Query OK, 0 rows affected (0,09 sec)

mysql> insert into test (`int_id`, `float`) VALUES (1,2);
Query OK, 1 row affected (0,00 sec)

mysql> select * from test;
+------+----------+-----+----------+
| int_id | int_nullable | float | float_nullable |
+------+----------+-----+----------+
| 1 | NULL | 2 | NULL |
+------+----------+-----+----------+
1 row in set (0,00 sec) 

   在CH服务端创建表，并从中查询数据：
CREATE TABLE jdbc_table
(
`int_id` Int32,
`int_nullable` Nullable(Int32),
`float` Float32,
`float_nullable` Nullable(Float32)
)
ENGINE JDBC('jdbc:mysql://localhost:3306/?user=root&password=root', 'test', 'test')

SELECT *
FROM jdbc_table;

┌─int_id─┬─int_nullable─┬─float─┬─float_nullable─┐
│ 1 │ ᴺᵁᴸᴸ │ 2 │ ᴺᵁᴸᴸ │
└────────┴──────────────┴───────┴────────────────┘   

   
   